package io.kiki.sba.registry.server.data_store.lease;


import io.kiki.sba.registry.api.Server;
import io.kiki.sba.registry.api.exchange.DataExchanger;
import io.kiki.sba.registry.common.model.ClientInterfaceId;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.slot.SlotConfig;
import io.kiki.sba.registry.common.model.store.ClientInterfaceIdIntern;
import io.kiki.sba.registry.remoting.netty.Channel;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.CleanContinues;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.change.DataChangeEventCenter;
import io.kiki.sba.registry.server.data_store.change.DataChangeType;
import io.kiki.sba.registry.server.data_store.slot.SlotManager;
import io.kiki.sba.registry.server.shared.meta.MetaServerService;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.LoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;


public final class ClientInterfaceLeaseManager {
    private static final Logger logger = LoggerFactory.getLogger(ClientInterfaceLeaseManager.class);


    private static final int MIN_LEASE_SEC = 5;
    private final Map<ClientInterfaceId, Long> clientInterfaceIdToLeaseMap = new ConcurrentHashMap<>();
    @Autowired
    ServerConfig serverConfig;
    @Resource
    DatumStorageDelegate datumStorageDelegate;
    @Autowired
    DataExchanger dataExchanger;
    @Autowired
    SlotManager slotManager;
    @Autowired
    MetaServerService metaServerService;
    @Autowired
    DataChangeEventCenter dataChangeEventCenter;

    public static void validateSessionLeaseSec(int sec) {
        if (sec < MIN_LEASE_SEC) {
            throw new IllegalArgumentException(String.format("min sessionLeaseSec(%d): %d", MIN_LEASE_SEC, sec));
        }
    }

    @PostConstruct
    public void init() {
        validateSessionLeaseSec(serverConfig.getSessionLeaseSecs());
        ConcurrentUtils.createDaemonThread("session-lease-cleaner", new LeaseCleaner()).start();
    }

    public void renewal(ClientInterfaceId clientInterfaceId) {
        clientInterfaceId = ClientInterfaceIdIntern.intern(clientInterfaceId);
        clientInterfaceIdToLeaseMap.put(clientInterfaceId, System.currentTimeMillis());
    }

    private List<ClientInterfaceId> cleanExpireLease(int leaseMs, Set<ClientInterfaceId> connClientInterfaceIds, Set<ClientInterfaceId> clientInterfaceIdSet) {
        final long lastRenew = System.currentTimeMillis() - leaseMs;
        Map<ClientInterfaceId, Long> map = new HashMap<>();
        for (Map.Entry<ClientInterfaceId, Long> entry : clientInterfaceIdToLeaseMap.entrySet()) {
            if (entry.getValue() < lastRenew) {
                map.put(entry.getKey(), entry.getValue());
            }
        }
        if (map.isEmpty()) {
            return Collections.emptyList();
        }

        logger.info("collectExpire={}, {}", map.size(), map);
        List<ClientInterfaceId> list = new ArrayList<>(map.size());
        for (Map.Entry<ClientInterfaceId, Long> entry : map.entrySet()) {
            final ClientInterfaceId clientInterfaceId = entry.getKey();
            if (connClientInterfaceIds.contains(clientInterfaceId)) {
                renewal(clientInterfaceId);
                // maybe happens in fullGc or high start scenario
                logger.info("[expireHasConn]{}", entry);
                continue;
            }
            if (clientInterfaceIdSet.contains(clientInterfaceId)) {
                renewal(clientInterfaceId);
                // maybe happens in fullGc or or high start scenario
                logger.info("[expireHasMeta]{}", entry);
                continue;
            }
            // only remove the expire with the collect timestamp
            // maybe renew happens in the cleaning
            if (clientInterfaceIdToLeaseMap.remove(clientInterfaceId, entry.getValue())) {
                logger.info("[cleanExpire]{}", entry);
                list.add(clientInterfaceId);
            }
        }
        return list;
    }

    void cleanStorage() {
        // make sure the existing processId be clean
        String dataCenter = serverConfig.getDataCenter();
        Set<ClientInterfaceId> clientInterfaceIdSet = datumStorageDelegate.getClientInterfaceIdSet(dataCenter);
        final int deadlineMillis = serverConfig.getSessionLeaseCleanDeadlineSecs() * 1000;
        for (int i = 0; i < SlotConfig.SLOT_NUM; i++) {
            if (slotManager.isFollower(dataCenter, i)) {
                logger.info("skip clean for follower, slotId={}", i);
                continue;
            }
            if (slotManager.isLeader(dataCenter, i)) {
                // own the slot and is leader
                for (ClientInterfaceId clientInterfaceId : clientInterfaceIdSet) {
                    if (!clientInterfaceIdToLeaseMap.containsKey(clientInterfaceId)) {
                        // double check with newly meta.processIds
                        long start = System.currentTimeMillis();
                        if (metaServerService.getClientInterfaceIdSet().contains(clientInterfaceId)) {
                            logger.info("[expireHasMeta]{}", clientInterfaceId);
                            continue;
                        }
                        logger.info("[cleanSession]{}", clientInterfaceId);
                        // in fullGC case, the cleaning maybe very slow
                        // after double check, the session.processId is insert, but cleanup not know it
                        // deadlineTs, ensure that the cleanup ends within the expected time
                        final long deadlineTimestamp = start + deadlineMillis;
                        CleanLeaseContinues cleanLeaseContinues = new CleanLeaseContinues(deadlineTimestamp);
                        Map<String, DatumVersion> versionMap = datumStorageDelegate.cleanByClientInterfaceId(dataCenter, i, clientInterfaceId, cleanLeaseContinues);
                        if (!versionMap.isEmpty()) {
                            dataChangeEventCenter.onChange(versionMap.keySet(), DataChangeType.lease, dataCenter);
                            logger.info("[cleanDatum]broken={},slotId={},datas={},pubs={},span={},{}", cleanLeaseContinues.broken, i, versionMap.size(), cleanLeaseContinues.cleanNum, System.currentTimeMillis() - start, clientInterfaceId);
                        }
                    }
                }
            }
        }
    }

    private void renewal(Set<ClientInterfaceId> clientInterfaceIdSet) {
        for (ClientInterfaceId clientInterfaceId : clientInterfaceIdSet) {
            renewal(clientInterfaceId);
        }
    }

    Set<ClientInterfaceId> getClientInterfaceIdSet() {
        Server server = dataExchanger.getServer(serverConfig.getPort());
        if (server == null) {
            logger.warn("Server not init when check session lease");
            return Collections.emptySet();
        }
        List<io.kiki.sba.registry.api.Channel> channelList = server.getChannelList();
        Set<ClientInterfaceId> set = new HashSet<>(128);
        for (io.kiki.sba.registry.api.Channel channel : channelList) {
            Channel channel1 = (Channel) channel;
            ClientInterfaceId clientInterfaceId = (ClientInterfaceId) channel1.getChannel().getAttribute(ValueConstants.ttr_rpc_channel_client_interface_id);
            if (clientInterfaceId != null) {
                set.add(clientInterfaceId);
            }
        }
        logger.info("find processId from channels={}, {}:{}", channelList.size(), set.size(), set);
        return set;
    }

    void clean() {
        // 1. renew the lease by conn and meta
        Set<ClientInterfaceId> clientInterfaceIdSet = getClientInterfaceIdSet();
        renewal(clientInterfaceIdSet);
        Set<ClientInterfaceId> metaClientInterfaceIds = metaServerService.getClientInterfaceIdSet();
        renewal(metaClientInterfaceIds);
        // 2. update the processId again, if has full gc,
        clientInterfaceIdSet = getClientInterfaceIdSet();
        metaClientInterfaceIds = metaServerService.getClientInterfaceIdSet();

        cleanExpireLease(serverConfig.getSessionLeaseSecs() * 1000, clientInterfaceIdSet, metaClientInterfaceIds);
        cleanStorage();

        // compact the unpub
        long tombstoneTimestamp = System.currentTimeMillis() - serverConfig.getDatumCompactDelaySecs() * 1000;
        String dataCenter = serverConfig.getDataCenter();
        Map<String, Integer> compacted = datumStorageDelegate.compact(dataCenter, tombstoneTimestamp);
        logger.info("compact datum, {}", compacted);
    }

    static final class CleanLeaseContinues implements CleanContinues {
        private final long deadlineTimestamp;
        volatile boolean broken;
        private int cleanNum;

        CleanLeaseContinues(long deadlineTimestamp) {
            this.deadlineTimestamp = deadlineTimestamp;
        }

        @Override
        public boolean continues() {
            final long now = System.currentTimeMillis();
            // make sure at lease clean one item
            if (cleanNum != 0 && now > deadlineTimestamp) {
                this.broken = true;
                return false;
            }
            return true;
        }

        @Override
        public void onClean(int num) {
            cleanNum += num;
        }
    }

    private final class LeaseCleaner extends LoopExecuteTask {
        @Override
        public void _execute_() {
            clean();
        }

        @Override
        public void _wait_() {
            // CPU overhead is high
            ConcurrentUtils.sleepUninterruptibly(serverConfig.getSessionLeaseCheckIntervalSecs(), TimeUnit.SECONDS);
        }
    }
}
